package com.reactor.demo;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

import java.awt.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * @author jing
 * @description
 * @date 2018/12/6 0006 22:02
 */

@RunWith(SpringRunner.class)
@SpringBootTest
public class ReactorDemo {
    @Test
    public void testStepVerifier(){
        StepVerifier.create(Flux.just(1,2,3,4))
                .expectNext(1,2,3,4)//用于测试下一个期望的数据元素,可以设置一些预期查看流的执行过程是否达到预期值
                .expectComplete()//用于测试下一个元素是否为完成信号
                .verify();

       /* StepVerifier.create(Mono.error(new Exception("this error")))
                .expectErrorMessage("receive the error")//用于校验下一个元素是否为错误信号
                .verify();*/

        StepVerifier.create(Flux.range(1,6)//表示从1开始自增
                            .map(i -> {
                                int a = i*i;
                                System.out.println("this is "+a);
                                return a;
                            }))
                            .expectNext(1, 4, 9, 16,25, 36)
                            .expectComplete()
                            .verify();//不写该方法好像一直没有执行完毕
    }

    /**
     * 测试mono和flux的订阅(subscrib)与发布(just)
     */
    @Test
    public void testJustAndSubscibe(){
        Integer[] array = new Integer[]{1,2,3,4,5,6};
        Flux.just("hehe","haha").subscribe(System.out::println);//一个Flux对象代表包含0...n个元素的响应式队列
        Mono.just("test").subscribe(System.out::println);//一个Mono对象包含1个包含0-1个元素的响应结果
        Flux.fromArray(array).subscribe(System.out::print);//创建一个参数为数组元素的响应式流
        System.out.println("");
        Flux.fromIterable(Arrays.asList(array)).subscribe(System.out::print);//创建一个参数为list元素的响应式流
    }

    /**
     * 将多个数据流合并成一个大的数据流
     */
    @Test
    public void testFlatMap(){
        StepVerifier.create(
                Flux.just("jing", "test").flatMap(s -> Flux.just(s + "demo"))
                        .delayElements(Duration.ofMillis(100))
                        .doOnNext(System.out::println).log("this is me log"))
                .expectNext("jingdemo", "testdemo")
                .verifyComplete();
    }

    /**
     * 响应式流可以发出三种数据信号，1、处理元素值；2、错误信号；3、完成信号
     * 其中错误信号和完成信号都是终止信号不能同时共存，完成信号告诉下游订阅者数据流正常结束，
     * 错误信号终止数据流的通知将错误传递给下游订阅者
     */
    @Test
    public void testSubscribe(){
        Flux.just(1,2,3,4,5).subscribe(
                System.out::println,//处理元素值
                System.err::println,//错误信号
                () -> System.out.println("Completed!"));//完成信号

        Flux.error(new Exception("some error")).subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("Completed!")
        );
    }

    /**
     * 测试过滤器filter
     */
    @Test
    public void testFilter(){
        StepVerifier.create(
                Flux.range(1,6)
                .filter(i -> i % 2 ==1)//过滤掉偶数只留奇数
                .map(i -> {
                    int a = i*i;
                    System.out.println(a);
                    return a;
                }))
                .expectNext(1,9,25)
                .verifyComplete();
    }

    private Flux<String> getTest() {
        String test = "this is jing";
        return Flux.fromArray(test.split("\\s+"));
    }

    /**
     * 测试flux的zip
     * @description (1)zip能够将多个流一对一的合并起来。它会每次从多个流中各取一个元素并将其合并为一
     *              (2)定义一个CountDownLatch，初始为1，则会等待执行1次countDown方法后结束，
     *              不使用它的话，测试方法所在的线程会直接返回而不会等待数据流发出完毕
     *              (3)总结flatMap是类似于将一个多层嵌套的流压缩成一个流，而zip类似于将多个流合并成一个大流
     * @throws InterruptedException
     */
    @Test
    public void testZip() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Flux.zip(
                getTest(),
                Flux.interval(Duration.ofMillis(200)))
                .subscribe(t -> System.out.println(t.getT1()), null, countDownLatch::countDown);
        countDownLatch.await(10, TimeUnit.SECONDS);//将线程挂起直到CountDownLatch的参数值为0时继续执行
    }

    public String getStringSync(){
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("当前线程号："+Thread.currentThread().getName());
        return "test test test";
    }

    @Test
    public void testSyncToAsync() throws InterruptedException {
        System.out.println("当前线程号："+Thread.currentThread().getName());
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Mono.fromCallable(() -> getStringSync())
                .subscribeOn(Schedulers.elastic())//使用subscribeOn将任务调度到Schedulers内置的弹性线程池执行
                .subscribe(System.out::println, null, countDownLatch::countDown);
        countDownLatch.await(11, TimeUnit.SECONDS);
    }

    @Test
    public void testErrorHandle(){
        Flux.range(1,5)
                .map(i -> 10/(i-3))
                .onErrorReturn(4)//onErrorReturn方法能够在收到错误信号的时候提供一个缺省值
                .map(i -> i*i)
                .subscribe(System.out::println, System.err::println);

        Flux.range(1, 5)
                .map(i -> 10 / (i - 3))
                .onErrorResume(e -> Mono.just(new Random().nextInt(6)))//onErrorResume方法能够在收到错误信号的时候提供一个新的数据流
                .map(i -> i * i)
                .subscribe(System.out::println);

    }

    /**
     *这个final不是太理解，不知道主要起了什么作用
     */
    @Test
    public void testFinal(){
        LongAdder longAdder = new LongAdder();
        Flux<String> flux = Flux.just("hehe", "haha")
                .doFinally(type -> {
                    System.out.println(type);
                    if(type == SignalType.CANCEL)
                        longAdder.increment();
                })
                .take(1);

        flux.subscribe(System.out::println);
    }

}
